package device

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"regexp"
	"sync"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/ev-charging/cloud-platform/config"
	"github.com/ev-charging/cloud-platform/internal/messaging"
)

// Device 设备结构体
type Device struct {
	ID              string             `json:"id"`
	Status          string             `json:"status"` // online, offline, charging, fault
	Model           string             `json:"model"`
	FirmwareVersion string             `json:"firmware_version"`
	LastActive      time.Time          `json:"last_active"`
	IPAddress       string             `json:"ip_address"`
	Location        map[string]string  `json:"location"` // latitude, longitude, address
	Metrics         map[string]float64 `json:"metrics"`  // current, voltage, power
}

// Service 设备服务
type Service struct {
	mqttClient      mqtt.Client
	kafkaProducer   *messaging.KafkaProducer
	config          *config.MQTTConfig
	db              map[string]*Device // 内存设备列表，实际项目中应使用分布式缓存
	mutex           sync.RWMutex
	deviceDataTopic string
	commandTopic    string
	ctx             context.Context
	cancel          context.CancelFunc
}

// NewService 创建设备服务实例
func NewService(cfg config.MQTTConfig, kafkaProducer *messaging.KafkaProducer) *Service {
	ctx, cancel := context.WithCancel(context.Background())
	return &Service{
		config:          &cfg,
		kafkaProducer:   kafkaProducer,
		db:              make(map[string]*Device),
		deviceDataTopic: cfg.Topics.DeviceTelemetry,
		commandTopic:    cfg.Topics.DeviceCommands,
		ctx:             ctx,
		cancel:          cancel,
	}
}

// Start 启动设备服务
func (s *Service) Start() error {
	// 初始化MQTT客户端
	opts := mqtt.NewClientOptions()
	opts.AddBroker(s.config.Broker)
	opts.SetClientID(s.config.ClientID)
	opts.SetUsername(s.config.Username)
	opts.SetPassword(s.config.Password)
	opts.SetCleanSession(s.config.CleanSession)

	// 设置连接回调
	opts.SetOnConnectHandler(s.onConnect)
	opts.SetConnectionLostHandler(s.onConnectionLost)

	// 设置心跳间隔
	keepAlive, err := time.ParseDuration(s.config.KeepAlive)
	if err != nil {
		return fmt.Errorf("解析心跳间隔失败: %w", err)
	}
	opts.SetKeepAlive(keepAlive)

	// 创建客户端
	s.mqttClient = mqtt.NewClient(opts)

	// 连接到MQTT broker
	if token := s.mqttClient.Connect(); token.Wait() && token.Error() != nil {
		return fmt.Errorf("MQTT连接失败: %w", token.Error())
	}

	log.Println("设备服务启动成功")

	// 启动设备状态检查协程
	go s.checkDeviceStatus()

	return nil
}

// onConnect 连接成功回调
func (s *Service) onConnect(client mqtt.Client) {
	log.Println("MQTT连接成功，开始订阅设备主题")

	// 订阅设备遥测数据主题
	token := client.Subscribe(s.deviceDataTopic, 1, s.handleDeviceData)
	token.Wait()
	if err := token.Error(); err != nil {
		log.Printf("订阅设备遥测主题失败: %v", err)
		return
	}

	log.Printf("已订阅设备遥测主题: %s", s.deviceDataTopic)
}

// onConnectionLost 连接丢失回调
func (s *Service) onConnectionLost(client mqtt.Client, err error) {
	log.Printf("MQTT连接丢失: %v, 尝试重连...", err)

	// 重连逻辑
	for {
		if token := client.Connect(); token.Wait() && token.Error() == nil {
			log.Println("MQTT重连成功")
			s.onConnect(client)
			return
		}

		log.Println("重连失败，5秒后重试...")
		time.Sleep(5 * time.Second)
	}
}

// handleDeviceData 处理设备遥测数据
func (s *Service) handleDeviceData(client mqtt.Client, msg mqtt.Message) {
	// 从主题中提取设备ID (主题格式: devices/{deviceID}/telemetry)
	re := regexp.MustCompile(`devices/(.*)/telemetry`)
	matches := re.FindStringSubmatch(msg.Topic())
	if len(matches) < 2 {
		log.Printf("无效的设备主题格式: %s", msg.Topic())
		return
	}
	deviceID := matches[1]

	// 解析设备数据
	var data map[string]interface{}
	if err := json.Unmarshal(msg.Payload(), &data); err != nil {
		log.Printf("解析设备数据失败: %v, 原始数据: %s", err, string(msg.Payload()))
		return
	}

	// 更新设备状态
	s.updateDeviceStatus(deviceID, data)

	// 将设备数据发送到Kafka
	go s.sendToKafka(deviceID, data)
}

// updateDeviceStatus 更新设备状态
func (s *Service) updateDeviceStatus(deviceID string, data map[string]interface{}) {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	device, exists := s.db[deviceID]
	if !exists {
		// 新设备，创建记录
		device = &Device{
			ID:       deviceID,
			Status:   "online",
			Metrics:  make(map[string]float64),
			Location: make(map[string]string),
		}

		// 提取设备基本信息（如果有）
		if model, ok := data["model"].(string); ok {
			device.Model = model
		}
		if fwVersion, ok := data["firmware_version"].(string); ok {
			device.FirmwareVersion = fwVersion
		}
		if ip, ok := data["ip_address"].(string); ok {
			device.IPAddress = ip
		}

		s.db[deviceID] = device
	}

	// 更新最后活动时间
	now := time.Now()
	device.LastActive = now

	// 更新设备状态
	if status, ok := data["status"].(string); ok {
		switch status {
		case "charging", "idle", "fault", "offline":
			device.Status = status
		}
	}

	// 更新 metrics
	if current, ok := data["current"].(float64); ok {
		device.Metrics["current"] = current
	}
	if voltage, ok := data["voltage"].(float64); ok {
		device.Metrics["voltage"] = voltage
	}
	if power, ok := data["power"].(float64); ok {
		device.Metrics["power"] = power
	}

	// 更新位置信息
	if lat, ok := data["latitude"].(string); ok {
		device.Location["latitude"] = lat
	}
	if lng, ok := data["longitude"].(string); ok {
		device.Location["longitude"] = lng
	}
}

// sendToKafka 将设备数据发送到Kafka
func (s *Service) sendToKafka(deviceID string, data map[string]interface{}) {
	if s.kafkaProducer == nil {
		log.Println("Kafka生产者未初始化，无法发送设备数据")
		return
	}

	// 添加设备ID和时间戳
	data["device_id"] = deviceID
	data["timestamp"] = time.Now().UnixMilli()

	// 转换为JSON
	jsonData, err := json.Marshal(data)
	if err != nil {
		log.Printf("设备数据JSON序列化失败: %v", err)
		return
	}

	// 发送到Kafka
	_, _, err = s.kafkaProducer.SendMessage("device-data", deviceID, string(jsonData))
	if err != nil {
		log.Printf("发送设备数据到Kafka失败: %v", err)
	}
}

// SendCommand 发送命令到设备
func (s *Service) SendCommand(deviceID string, command map[string]interface{}) error {
	if !s.mqttClient.IsConnected() {
		return fmt.Errorf("MQTT客户端未连接")
	}

	// 添加命令ID和时间戳
	command["command_id"] = fmt.Sprintf("cmd-%d", time.Now().UnixNano())
	command["timestamp"] = time.Now().UnixMilli()

	// 转换为JSON
	jsonCmd, err := json.Marshal(command)
	if err != nil {
		return fmt.Errorf("命令JSON序列化失败: %w", err)
	}

	// 发送命令到设备特定主题
	topic := fmt.Sprintf("devices/%s/commands", deviceID)
	token := s.mqttClient.Publish(topic, 1, false, jsonCmd)
	token.Wait()

	if err := token.Error(); err != nil {
		return fmt.Errorf("发送命令失败: %w", err)
	}

	log.Printf("已发送命令到设备 %s: %s", deviceID, string(jsonCmd))
	return nil
}

// checkDeviceStatus 检查设备状态，标记离线设备
func (s *Service) checkDeviceStatus() {
	// 从配置获取心跳间隔
	interval, err := time.ParseDuration("30s") // 实际项目中应从配置获取
	if err != nil {
		log.Printf("解析心跳间隔失败，使用默认值30秒: %v", err)
		interval = 30 * time.Second
	}

	// 定期检查
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			s.mutex.Lock()
			now := time.Now()
			offlineThreshold := now.Add(-2 * interval) // 超过2个心跳周期未活动则标记为离线

			for id, device := range s.db {
				if device.Status != "offline" && device.LastActive.Before(offlineThreshold) {
					device.Status = "offline"
					log.Printf("设备 %s 已离线", id)

					// 发送设备离线事件到Kafka
					go s.sendDeviceEvent(id, "offline")
				}
			}
			s.mutex.Unlock()

		case <-s.ctx.Done():
			log.Println("设备状态检查协程退出")
			return
		}
	}
}

// sendDeviceEvent 发送设备事件到Kafka
func (s *Service) sendDeviceEvent(deviceID, eventType string) {
	if s.kafkaProducer == nil {
		return
	}

	event := map[string]interface{}{
		"device_id":  deviceID,
		"event_type": eventType,
		"timestamp":  time.Now().UnixMilli(),
	}

	jsonEvent, err := json.Marshal(event)
	if err != nil {
		log.Printf("设备事件JSON序列化失败: %v", err)
		return
	}

	_, _, err = s.kafkaProducer.SendMessage("device-events", deviceID, string(jsonEvent))
	if err != nil {
		log.Printf("发送设备事件到Kafka失败: %v", err)
	}
}

// GetDevice 获取设备信息
func (s *Service) GetDevice(deviceID string) (*Device, bool) {
	s.mutex.RLock()
	defer s.mutex.RUnlock()

	device, exists := s.db[deviceID]
	return device, exists
}

// ListDevices 列出所有设备
func (s *Service) ListDevices() []*Device {
	s.mutex.RLock()
	defer s.mutex.RUnlock()

	devices := make([]*Device, 0, len(s.db))
	for _, device := range s.db {
		devices = append(devices, device)
	}

	return devices
}

// Stop 停止设备服务
func (s *Service) Stop() {
	if s.cancel != nil {
		s.cancel() // 取消上下文
	}

	if s.mqttClient != nil && s.mqttClient.IsConnected() {
		s.mqttClient.Disconnect(250) // 断开 MQTT 连接
		log.Println("MQTT客户端已断开连接")
	}

	log.Println("设备服务已停止")
}
